package jms;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;

public class BossWorkerPool<T> {
	
	private static final Logger LOGGER = Logger.getLogger(BossWorkerPool.class.getName());
	
	public static interface WorkCreator<T> {
		Runnable createWork(T product);
	}
	
	private final ExecutorService bossPool;
	private final ExecutorService workerPool;
	private final int maxWorker;
	private final WorkCreator<T> workCreator;
	private final AtomicBoolean started = new AtomicBoolean(false);
	private final AtomicInteger workerCounter = new AtomicInteger(0);
	private final BlockingQueue<T> queue;
	
	private BossWorkerPool(WorkCreator<T> workCreator, BlockingQueue<T> queue, 
			int maxWorker, String namePrefix) {
		if(maxWorker <= 0)
			throw new IllegalArgumentException("max worker " + maxWorker);
		if(workCreator == null)
			throw new NullPointerException("work creator");
		if(queue == null)
			throw new NullPointerException("queue");
		
		this.bossPool = 
				Executors.newSingleThreadExecutor(new NamedThreadFactory(namePrefix + "-Boss"));
		this.workerPool =
				Executors.newCachedThreadPool(new NamedThreadFactory(namePrefix + "-Worker"));
		this.maxWorker = maxWorker; 		
		this.workCreator = workCreator;
		this.queue = queue;
	}
	
	private void start() {
		if(!started.compareAndSet(false, true))
			throw new IllegalStateException("Started");
			
		bossPool.submit(new Runnable() {
			
			public void run() {
				while(true) {
					final T product;
					
					try {
						product = queue.take();
					} catch(InterruptedException ex) {
						LOGGER.info("Interrupted");
						return;	//interruption handler for shutdownNow() request
					}
					
					while(workerCounter.incrementAndGet() >= maxWorker) {
						workerCounter.decrementAndGet();
						try {
							Thread.sleep(100);	//sleep for a short time to relief the CPU
						} catch(InterruptedException ex) {
							LOGGER.info("Interrupted");
							return;
						}						
					}
					
					workerPool.execute(new Runnable() {
						
						public void run() {
							try {
								workCreator.createWork(product).run();
							} finally {
								workerCounter.decrementAndGet();
							}
						}
					});
				}
			}
		});
	}
	
	public void shutdownNow() {
		bossPool.shutdownNow();
		workerPool.shutdown();
		try {
			workerPool.awaitTermination(10, TimeUnit.SECONDS);
		} catch(InterruptedException ex) {
			Thread.currentThread().interrupt();
		}
		workerPool.shutdownNow();
	}
	
	public final static class Builder<T> {
		private int maxWorker;
		private WorkCreator<T> workCreator;
		private BlockingQueue<T> queue;		
		private String namePrefix = "";
		
		public Builder<T> setMaxWorker(int maxWorker) {
			this.maxWorker = maxWorker;
			return this;
		}
		
		public Builder<T> setNamePrefix(String namePrefix) {
			this.namePrefix = namePrefix;
			return this;
		}
		
		public Builder<T> setQueue(BlockingQueue<T> queue) {
			this.queue = queue;
			return this;
		}
		
		public Builder<T> setWorkCreator(WorkCreator<T> workCreator) {
			this.workCreator = workCreator;
			return this;
		}
		
		public BossWorkerPool<T> buildAndStart() {
			if(namePrefix == null) {
				namePrefix = "";
			}
			
			BossWorkerPool<T> pool = 
					new BossWorkerPool<T>(workCreator, queue, maxWorker, namePrefix);
			pool.start();
			return pool;
		}
	}
	
	

}
